﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Collections.Concurrent;
using System.ComponentModel;
using System.Threading;
using System.Threading.Tasks;
using System.Diagnostics;
using System.Transactions;
using MessageRouter.Configuration;

namespace MessageRouter
{
    public abstract class MessageReader : IObservable<RoutedMessage>
    {
        private readonly string name;
        private readonly Uri address;

        private readonly ConcurrentDictionary<Guid, MessageReaderSubscriber> subscribersDictionary;
        private readonly BackgroundWorker messageReaderWorker;
        private readonly AutoResetEvent cancelationPending;
        private readonly AutoResetEvent activationCompleted;

        public void Open()
        {
            this.OnOpening();
            this.messageReaderWorker.RunWorkerAsync();
            this.OnOpened();
        }

        public void Close()
        {
            this.OnClosing();
            this.cancelationPending.Set();
            this.messageReaderWorker.CancelAsync();
            this.OnClosed();
        }

        protected MessageReader(ReaderConfigurationElement config)
        {
            if (config == null)
            {
                throw new ArgumentNullException("config");
            }

            this.name = config.Name;
            this.address = config.Address;

            this.subscribersDictionary = new ConcurrentDictionary<Guid, MessageReaderSubscriber>();

            this.messageReaderWorker = new BackgroundWorker();
            this.messageReaderWorker.DoWork += MessageReaderWorkerProc;
            this.messageReaderWorker.WorkerSupportsCancellation = true;

            this.cancelationPending = new AutoResetEvent(false);
            this.activationCompleted = new AutoResetEvent(false);
        }

        public string Name
        {
            get { return this.name; }
        }

        public Uri Address
        {
            get { return this.address; }
        }

        /// <summary>
        /// Read next message from queue
        /// </summary>
        /// <returns></returns>
        protected abstract bool TryRead(out RoutedMessage routedMessage);

        /// <summary>
        /// Notified reader about activation completed succesfully
        /// </summary>
        protected void OnActivationCompleted()
        {
            this.activationCompleted.Set();
        }

        private void SubscriberDisposed(object sender, EventArgs e)
        {
            MessageReaderSubscriber subscriber = sender as MessageReaderSubscriber;
            if (subscriber == null) throw new ArgumentException("sender");
            subscriber.Disposed -= SubscriberDisposed;
            this.subscribersDictionary.TryRemove(subscriber.Id, out subscriber);
        }

        protected virtual void MessageReaderWorkerProc(object sender, DoWorkEventArgs e)
        {
            while (WaitHandle.WaitAny(new WaitHandle[] { cancelationPending, activationCompleted }) > 0)
            {
                try
                {
                    while (true)
                    {
                        using (TransactionScope transaction = new TransactionScope())
                        {
                            try
                            {
                                RoutedMessage message = null;
                                if (this.TryRead(out message))
                                {
                                    this.OnNext(message);
                                    transaction.Complete();
                                    continue;
                                }
                                else
                                {
                                    break;
                                }
                            }
                            catch (Exception ex)
                            {
                                this.OnError(ex);
                                break;
                            }
                        }
                    }
                }
                catch (Exception ex)
                {
                    this.OnError(ex);
                }
            }
            this.OnCompleted();
        }
        

        #region protected virtuals

        protected virtual void OnInitialize()
        {
        }

        /// <summary>
        /// MessageReader opening
        /// </summary>
        protected virtual void OnOpening()
        {
        }

        /// <summary>
        /// MessageReader opened
        /// </summary>
        protected virtual void OnOpened()
        {
        }

        /// <summary>
        /// MessageReader closing
        /// </summary>
        protected virtual void OnClosing()
        {
        }

        /// <summary>
        /// MessageReader closed
        /// </summary>
        protected virtual void OnClosed()
        {
        }

        /// <summary>
        /// Notified subscribers about next message
        /// </summary>
        /// <param name="message"></param>
        protected virtual void OnNext(RoutedMessage message)
        {
            MessageReaderSubscriber[] subscribers = this.subscribersDictionary.Values.ToArray();
            foreach (var subscriber in subscribers)
            {
                subscriber.Observer.OnNext(message);
            }
        }

        /// <summary>
        /// Notified subscribers about error
        /// </summary>
        /// <param name="error"></param>
        protected virtual void OnError(Exception error)
        {
            MessageReaderSubscriber[] subscribers = this.subscribersDictionary.Values.ToArray();
            foreach (var subscriber in subscribers)
            {
                subscriber.Observer.OnError(error);
            }
        }

        /// <summary>
        /// Notified subscribers about notification canceled
        /// </summary>
        protected virtual void OnCompleted()
        {
            MessageReaderSubscriber[] subscribers = this.subscribersDictionary.Values.ToArray();
            foreach (var subscriber in subscribers)
            {
                subscriber.Observer.OnCompleted();
            }
        }

        #endregion

        #region IObservable<RouterMessage> interface implementation

        public IDisposable Subscribe(IObserver<RoutedMessage> observer)
        {
            var subscriber = new MessageReaderSubscriber(observer);
            subscriber.Disposed += SubscriberDisposed;
            this.subscribersDictionary.TryAdd(subscriber.Id, subscriber);
            return subscriber;
        }

        public bool HasSubscribers
        {
            get { return !this.subscribersDictionary.IsEmpty; }
        }

        #endregion

        #region private sealed class MessageReaderSubscriber : IDisposable
        private sealed class MessageReaderSubscriber : IDisposable
        {
            private readonly Guid id;
            private readonly IObserver<RoutedMessage> observer;
            public MessageReaderSubscriber(IObserver<RoutedMessage> observer)
            {
                this.id = Guid.NewGuid();
                this.observer = observer;
            }

            public Guid Id
            {
                get
                {
                    return this.id;
                }
            }

            public IObserver<RoutedMessage> Observer
            {
                get
                {
                    return this.observer;
                }
            }

            public event EventHandler Disposed;
            public void Dispose()
            {
                var handle = this.Disposed;
                if (handle != null)
                {
                    handle(this, new EventArgs());
                }
            }
        }
        #endregion
    }
}
